package com.intct.util;

import org.apache.flink.connector.datagen.table.DataGenConnectorOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author gufg
 * @since 2025-06-30 17:16
 */
public class FlinkSQL2 {
    public static void main(String[] args) {
        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        TableDescriptor tableDescriptor = TableDescriptor.forConnector("datagen")
                .schema(Schema.newBuilder()
                        .column("c1", DataTypes.STRING())
                        .column("c2", DataTypes.STRING())
                        .column("c3", DataTypes.STRING())
                        .build()
                )
                .option(DataGenConnectorOptions.ROWS_PER_SECOND, 5L)
                .build();

        // 创建一个临时表
        tEnv.createTemporaryTable("test1", tableDescriptor);

        // 查询--Table API
        Table table = tEnv.from("test1");

        Table limit = table.limit(2);

        limit.execute().print();
//        result.print();

        // 查询-- SQL
//        TableResult res = tEnv.executeSql("select * from test1 limit 1");
//        res.print();

    }
}
